package com.iot.service.mqtt.handler;

import com.iot.service.mqtt.config.Constant;
import com.iot.service.mqtt.model.TlMqttSession;
import com.iot.service.mqtt.model.payload.TlMqttConnectPayload;
import com.iot.service.mqtt.model.request.TlMqttConnectReq;
import com.iot.service.mqtt.model.request.TlMqttPublishReq;
import com.iot.service.mqtt.model.response.TlMqttConnack;
import com.iot.service.mqtt.model.variable.TlMqttConnectVariableHead;
import com.iot.service.mqtt.service.auth.AuthenticationService;
import com.iot.service.mqtt.service.pub.IPublishStoreService;
import com.iot.service.mqtt.service.rel.IPubrelStoreService;
import com.iot.service.mqtt.service.retain.IRetainService;
import com.iot.service.mqtt.service.session.ISessionService;
import com.iot.service.mqtt.service.subscripe.ISubscriptionService;
import com.iot.service.mqtt.utils.ChannelManager;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.handler.codec.mqtt.MqttConnectReturnCode;
import io.netty.handler.codec.mqtt.MqttQoS;
import io.netty.handler.codec.mqtt.MqttVersion;
import io.netty.handler.timeout.IdleStateHandler;
import io.netty.util.AttributeKey;
import lombok.RequiredArgsConstructor;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

import java.util.HashSet;
import java.util.concurrent.TimeUnit;

/**
 * @ClassName：ConnectHandler
 * @Author: hszhou
 * @Date: 2024/11/25 11:06
 * @Description: 必须描述类做什么事情, 实现什么功能
 */
@ChannelHandler.Sharable
@Component
@RequiredArgsConstructor
public class TlMqttConnectHandler extends SimpleChannelInboundHandler<TlMqttConnectReq> {

    private static final Logger logger = LoggerFactory.getLogger(TlMqttConnectHandler.class);

    public final ISubscriptionService subscriptionService;

    private final ISessionService sessionService;

    private final IRetainService retainMsgService;

    private final IPublishStoreService publishStoreService;

    private final IPubrelStoreService pubrelStoreService;

    private final ChannelManager channelManager;

    private final AuthenticationService authenticationService;

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, TlMqttConnectReq req) throws Exception {
        logger.debug("in【CONNECT】 handler");
        Channel channel = ctx.channel();
        TlMqttConnectVariableHead variableHead = req.getVariableHead();
        TlMqttConnectPayload payload = req.getPayload();
        Short protocolVersion = variableHead.getProtocolVersion();
        //是否清除会话
        int cleanSession = variableHead.getCleanSession();
        String clientId = payload.getClientId();
        //协议不通过
        if (!validateProtocolVersion(protocolVersion, cleanSession, channel)) {
            return;
        }
        //连接标识  如果连接标识不是0 则断开连接
        int reserved = variableHead.getReserved();
        if (!validateReserved(reserved, channel)) {
            logger.error("客户端【{}】的连接标识不对，断开连接", clientId);
            return;
        }
        String username = payload.getUsername();
        String password = payload.getPassword();
        //认证
        boolean authentication = authentication(username, password);
        if(!authentication){
            TlMqttConnack res = TlMqttConnack.build(cleanSession, MqttConnectReturnCode.CONNECTION_REFUSED_BAD_USER_NAME_OR_PASSWORD);
            channel.writeAndFlush(res);
            return;
        }

        logger.debug("clientId is 【{}】", clientId);
        boolean sessionPresent = cleanSession != 0;
        //如果是1的话 就重新清除之前的会话
        if (sessionPresent) {
            actionOnCleanSession(clientId);
        }

        sessionService.find(clientId)
                /*
                 * 如果CleanSession被设置为0
                 * 服务器必须根据当前的会话状态恢复与客户端的通信（客户端的唯一标识作为会话的标识）。
                 * 如果没有与客户端唯一标识相关的绘画，服务端必须创建一个新的会话。
                 * 客户端和服务端在断开连接后必须存储会话[MQTT-3.1.2-4]
                 */
                .switchIfEmpty(Mono.defer(() -> Mono.just(TlMqttSession.build(clientId, sessionPresent))))
                .flatMap(session -> {
                    //将通道保存起来
                    channelManager.put(clientId, channel);
                    // 设置客户端标识以及正常断开连接的标识位为false 断开标志位设备false，表示没有走正常断开的逻辑 等到走正常关闭的逻辑时就可以设置为true
                    channel.attr(AttributeKey.valueOf(Constant.CLIENT_ID)).set(clientId);
                    channel.attr(AttributeKey.valueOf(Constant.DISCONNECT)).set(false);
                    channel.attr(AttributeKey.valueOf(Constant.USERNAME)).set(username);
                    channel.attr(AttributeKey.valueOf(Constant.IP)).set(null);
                    //构建一个connectAck消息
                    TlMqttConnack res = TlMqttConnack.build(cleanSession, MqttConnectReturnCode.CONNECTION_ACCEPTED);
                    channelManager.writeAndFlush(clientId, res);
                    logger.debug("send ack message to client 【{}】 connect  success", clientId);
                    if (!sessionPresent) {
                        republish(clientId);
                        session.setTopics(new HashSet<>());
                        session.setCleanSession(sessionPresent);
                    }
                    logger.info("client【{}】 connect success[{}]", clientId,channel);
                    return sessionService.save(session);
                }).subscribe(e -> {
                    //添加心跳检测
                    Short heartbeat = variableHead.getKeepAlive();
                    ctx.pipeline().addLast(new IdleStateHandler(0, 0, heartbeat, TimeUnit.SECONDS));
                });


        //保存遗嘱消息
        if (variableHead.getWillFlag() == 1) {
            saveWillMessage(payload.getWillTopic(), MqttQoS.valueOf(variableHead.getWillQos()),
                    variableHead.getWillRetain() != 0, payload.getWillMessage(), clientId)
                    .subscribe();
        }

    }

    /**
     * @description: 存储遗嘱消息
     * @author: hszhou
     * @datetime: 2025-04-28 16:48:35
     * @param:
     * @param: willTopic
     * @param: qoS
     * @param: isRetain
     * @param: messageContent
     * @param: clientId
     * @return: Mono<Boolean>
     **/
    private Mono<Boolean> saveWillMessage(String willTopic, MqttQoS qoS, boolean isRetain, String messageContent, String clientId) {
        TlMqttPublishReq message = TlMqttPublishReq.build(willTopic, qoS, isRetain, messageContent, 0L);
        Mono<Boolean> savePublisher = publishStoreService.saveWillMessage(clientId, message);
        Mono<Boolean> retainPublisher = isRetain ? retainMsgService.save(willTopic, message) : Mono.empty();
        return savePublisher.then(retainPublisher);
    }

    /**
     * @description: 校验连接标识
     * 服务端必须验证CONNECT控制包的预留字段是否为0，如果不为0断开与客户端的连接[MQTT-3.1.2-3].
     * @author: hszhou
     * @datetime: 2025-04-28 15:53:05
     * @param:
     * @param: reserved
     * @param: channel
     * @return: boolean
     **/
    private boolean validateReserved(int reserved, Channel channel) {
        if (reserved != 0) {
            channel.close();
            return false;
        }
        return true;
    }

    /**
     * @description: 清除客户端的信息
     * @author: hszhou
     * @datetime: 2025-04-28 16:32:19
     * @param:
     * @param: clientId 客户端ID
     * @return: void
     **/
    private void actionOnCleanSession(String clientId) {
        // sessionService.clear(String ClientId) 方法返回 true 则表明该 clientId 之前的 cleanSession = false, 那么应该继续清理
        // 订阅信息、pub 信息、 pubRel 信息, 否则无需清理
        sessionService.clear(clientId).flatMap(e -> {
            if (e) {
                return Mono.when(
                        //清除所有的订阅
                        subscriptionService.clear(clientId),
                        //清除保存的消息
                        publishStoreService.clearPublishMessage(clientId), pubrelStoreService.clearPubRelMessage(clientId),
                        //清除遗嘱消息
                        publishStoreService.clearWillMessage(clientId));
            } else {
                return Mono.empty();
            }
        }).subscribe();
    }


    private boolean validateProtocolVersion(Short protocolVersion, int cleanSession, Channel channel) {
        if (!(protocolVersion.toString().equals(MqttVersion.MQTT_5.protocolLevel() + "") || protocolVersion.toString().equals(MqttVersion.MQTT_3_1_1.protocolLevel() + ""))) {
            TlMqttConnack res = TlMqttConnack.build(cleanSession, MqttConnectReturnCode.CONNECTION_REFUSED_UNACCEPTABLE_PROTOCOL_VERSION);
            channel.writeAndFlush(res);
            return false;
        }
        return true;
    }

    /**
     * 客户端接入认证
     * @author hszhou
     * @datetime: 2025-05-10 16:21:33
     * @param username 用户名
     * @param password 密码
     * @return boolean
     **/
    private boolean authentication( String username, String password) {
        return authenticationService.auth(username,password);
    }



    /**
     * @description: 上线时补发qos1与qos2的消息
     * @author: hszhou
     * @datetime: 2025-05-08 15:09:57
     * @param:
     * @param: clientId
     * @return: void
     **/
    private void republish(String  clientId) {
        //补发qos1与qos的2的消息 未回复的消息
        logger.debug("cleanSession is 0 and republish message");
        Flux.merge(publishStoreService.findPublishMessage(clientId),
                pubrelStoreService.findPubRelMessage(clientId))
                .subscribe(e->{
                    logger.debug("republish message {}",e);
                    channelManager.writeAndFlush(clientId,e);
                });


    }
}
